package com.xukang.qd.boot;


import com.ibm.jms.JMSBytesMessage;
import com.ibm.jms.JMSMessage;
import com.ibm.jms.JMSTextMessage;
import com.ibm.mq.*;

import com.xukang.qd.listener.ConsumerIbmMqMessageListener;
import com.xukang.qd.boot.pojo.CcbMessage;
import org.apache.commons.codec.binary.Hex;



import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jms.JmsException;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.util.StringUtils;

import javax.jms.JMSException;
import javax.servlet.http.HttpServletRequest;
import java.io.UnsupportedEncodingException;


/**
 * @Description 消息队列工具类
 * @Author dming
 * @Time 2018-12-26
 */

public class MQUtils {

    private static final String correlationId = "REQ\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0";
    private static final String userID = "CHINATAX";
    private static final String type_sw_yh_online_s = "S_Y_O_S";//传输类型税务到银行同步 (请求)
    private static final String type_sw_yh_batch_s = "S_Y_B_S";//传输类型税务到银行异步 (请求)
    private static final String type_yh_sw_online_s = "Y_S_O_S";//传输类型银行到税务同步 (请求)
    private static final String type_yh_sw_batch_s = "Y_S_B_S";//传输类型银行到税务异步 (请求)



    @Value("${project.mq.syn-send-queue}")
    private String synSendQueue;
    @Value("${project.mq.syn-receive-queue}")
    private String synReceiveQueue;
    @Value("${project.mq.asyn-send-queue}")
    private String asynSendQueue;
    @Value("${project.mq.asyn-receive-queue}")
    private String asynReceiveQueue;
    @Value("${project.mq.send-success}")
    private String sendSuccess;
    @Value("${project.mq.send-fail}")
    private String sendFail;
    @Value("${project.mq.bbc-http-url}")
    private String BBC_HTTP_URL;
    @Value("${project.mq.host}")
    private String host;
    @Value("${project.mq.port}")
    private Integer port;
    @Value("${project.mq.queue-manager}")
    private String queueManager;
    @Value("${project.mq.channel}")
    private String channel;
    @Value("${project.mq.username}")
    private String username;
    @Value("${project.mq.password}")
    private String password;
    @Value("${project.mq.receive-timeout}")
    private long receiveTimeout;
    @Value("${project.mq.ccsid}")
    private int ccsid;


    public static Logger logger= LoggerFactory.getLogger(ConsumerIbmMqMessageListener.class);


    public static String getErrorXml(String errorMessage) {
        String errorXml = "<?xml version=\"1.0\" encoding=\"GBK\"?>\n" +
                "<tiripPackage version=\"1.0\" xmlns=\"http://www.chinatax.gov.cn/tirip/dataspec\">\n" +
                "<returnState>\n" +
                "<returnCode>F92</returnCode>\n" +
                "<returnMessage>" + errorMessage + "</returnMessage>\n" +
                "</returnState>\n" +
                "</tiripPackage>";
        return errorXml;
    }

    /**
     * 银行端请求发送消息，将消息写入消息队列.
     * 1.根据参数flag区分操作是同步还是异步，写入不同的消息队列中
     * 2.向远程消息队列中写入接收到银行端的xml，获取到messageid并将发送信息落地
     * 2.根据messageid，在接收队列中通过 correlationId = messageid 获取消息信息
     * 3.获取到消息信息则http调用银行端接口，将消息队列信息传递给银行端，并将返回信息更新步骤1中消息信息的response_xml
     * 4.接收不到目前未定，暂定为返回空字符串
     *
     * @param reqXml
     * @param flag
     * @return
     */
    public String sendMessage(String flag, String reqXml, HttpServletRequest request) {
        logger.error("==================== send message start ====================");
        logger.error("param : flag [" + flag + "] reqxml [" + reqXml + "]");
        String response_xml = "";
        try {
            //调用写入消息队列方法
            MQEnvironment.hostname = host;       //MQ服务器的IP地址
            MQEnvironment.channel = channel; //通道类型:服务器连接通道
            MQEnvironment.port = port;                     //MQ端口
            MQEnvironment.CCSID = ccsid;                      //服务器MQ服务使用的编码1381代表GBK、1208代表UTF(Coded Character Set Identifier:CCSID)
            MQQueueManager mqMgr = new MQQueueManager(queueManager); //设置 队列管理器序号

            int openOptions = MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING | MQC.MQOO_SET_IDENTITY_CONTEXT;
            /**
             * 关闭了就重新打开
             */
            if (mqMgr == null || !mqMgr.isConnected()) {
                mqMgr = new MQQueueManager(queueManager);
            }
            MQQueue queue;

            CcbMessage sendMessage = new CcbMessage();
            sendMessage.setStatus("0");
            sendMessage.setMessageSend(reqXml);
            String flagStr;
            if (flag.equalsIgnoreCase("syn")) {
                //同步发送队列
                queue = mqMgr.accessQueue(synSendQueue, openOptions);
                flagStr = type_yh_sw_online_s;
            } else {
                flagStr = type_yh_sw_batch_s;
                queue = mqMgr.accessQueue(asynSendQueue, openOptions);
            }
            sendMessage.setTypes(flagStr);

            //第一步：设置写入消息的属性
            MQPutMessageOptions pmo = new MQPutMessageOptions();
            pmo.options = pmo.options + MQC.MQPMO_NEW_MSG_ID;
            pmo.options = pmo.options + MQC.MQPMO_SYNCPOINT;
            pmo.options = pmo.options + MQC.MQPMO_SET_IDENTITY_CONTEXT;
            //第二部：定义消息写入缓冲区
            MQMessage putMessage = new MQMessage();

            putMessage.correlationId = correlationId.getBytes("GBK");
            putMessage.userId = userID;
            //将数据放入消息缓冲区
            putMessage.write(reqXml.getBytes("GBK"));
            //第三步：将消息写入队列
            queue.put(putMessage, pmo);
            //第四步：保存messageID,在接收MQ消息的时候设置correlationId为messageId
            mqMgr.commit();
            logger.error("send message success , messageid is [" + Hex.encodeHexString(putMessage.messageId) + "]");
            sendMessage.setMessageid(Hex.encodeHexString(putMessage.messageId));
            queue.close();
            mqMgr.disconnect();

            if (flag.equalsIgnoreCase("syn")) {
                //同步发送消息，发送成功后等待接收反应消息
                response_xml = this.receiveMessage (flag, sendMessage.getMessageid());
                sendMessage.setMessageReceive(response_xml);
                if (!StringUtils.isEmpty(response_xml)) {
                    sendMessage.setStatus("2");
                } else {
                    sendMessage.setStatus("1");
                }
            } else {
                //异步发送消息，发送成功后结束
                sendMessage.setMessageReceive(response_xml);
                sendMessage.setStatus("2");
            }

        } catch (JmsException jmsException) {
            logger.error("send message failed,exception message is [" + jmsException.getMessage() + "]");
        } catch (Exception e) {
            logger.error("send message failed,exception message is [" + e.getMessage() + "]");
        }
        if (!StringUtils.isEmpty(response_xml)) {
            String errorXml = this.getErrorXml("未获取到局端响应");
            logger.error("receive message is empty");
            logger.error("==================== send message end ====================");
            return errorXml;
        } else {
            logger.error("==================== send message end ====================");
            return response_xml;
        }

    }
    public String sendMessage(String flag, String reqXml, String messageID) {
        logger.error("send message after syn listener receive message , flag : [" + flag + "]" + ", responseXml : [" + reqXml + "]" +
                "message ID : [" + messageID + "]");
        String response_xml = "";
        try {
            //调用写入消息队列方法
            MQEnvironment.hostname = host;       //MQ服务器的IP地址
            MQEnvironment.channel = channel; //通道类型:服务器连接
            MQEnvironment.port = port;                     //MQ端口
            MQEnvironment.CCSID = ccsid;                      //服务器MQ服务使用的编码1381代表GBK、1208代表UTF(Coded Character Set Identifier:CCSID)
            MQQueueManager mqMgr = new MQQueueManager(queueManager);

            int openOptions = MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING | MQC.MQOO_SET_IDENTITY_CONTEXT;
            /**
             * 关闭了就重新打开
             */
            if (mqMgr == null || !mqMgr.isConnected()) {
                mqMgr = new MQQueueManager(queueManager);
            }
            MQQueue queue;

            if (flag.equalsIgnoreCase("syn")) {
                queue = mqMgr.accessQueue(synSendQueue, openOptions);
            } else {
                queue = mqMgr.accessQueue(asynSendQueue, openOptions);
            }

            //第一步：设置写入消息的属性
            MQPutMessageOptions pmo = new MQPutMessageOptions();
            pmo.options = pmo.options + MQC.MQPMO_NEW_MSG_ID;
            pmo.options = pmo.options + MQC.MQPMO_SYNCPOINT;
            pmo.options = pmo.options + MQC.MQPMO_SET_IDENTITY_CONTEXT;
            //第二部：定义消息写入缓冲区
            MQMessage putMessage = new MQMessage();

            if (flag.equalsIgnoreCase("syn")) {
                putMessage.correlationId = Hex.decodeHex(messageID.toCharArray());
            } else {
                putMessage.correlationId = messageID.getBytes("GBK");
            }

            putMessage.userId = userID;
            //将数据放入消息缓冲区
            putMessage.write(reqXml.getBytes("GBK"));
            //第三步：将消息写入队列
            queue.put(putMessage, pmo);
            //第四步：保存messageID,在接收MQ消息的时候设置correlationId为messageId
            mqMgr.commit();
            logger.error("send message after " + flag + " listener receive message [success]");
            queue.close();
            mqMgr.disconnect();

        } catch (JmsException jmsException) {
            logger.error("send message after " + flag + " listener receive message [failed] [" + jmsException.getMessage() + "]");
        } catch (Exception e) {
            logger.error("send message after " + flag + " listener receive message [failed] [" + e.getMessage() + "]");
        }
        logger.error("send message after " + flag + " listener receive message [end]");
        return response_xml;
    }

    /**
     * 根据flag  判断从哪个消息队列中获取消息
     * 根据 correlationId = messageid 获取具体哪儿一个消息
     *
     * @param flag
     * @param messageID
     * @return
     */
    public String receiveMessage(String flag, String messageID) {
        logger.error("receive tax response after send success , message ID [" + messageID + "]");
        String response_xml;
        try {
            MQQueueManager mqMgr = new MQQueueManager(queueManager);

            /**
             * 关闭了就重新打开
             */
            if (mqMgr == null || !mqMgr.isConnected()) {
                mqMgr = new MQQueueManager(queueManager);
            }
            MQQueue queue;
            String receiveStr;
            if (flag.equalsIgnoreCase("syn")) {
                receiveStr = type_yh_sw_online_s;
                queue = mqMgr.accessQueue(synReceiveQueue, MQC.MQGMO_WAIT, null, null, null);

            } else {
                receiveStr = type_yh_sw_batch_s;
                queue = mqMgr.accessQueue(asynReceiveQueue, MQC.MQGMO_WAIT, null, null, null);
            }

            MQGetMessageOptions gmo = new MQGetMessageOptions();
            gmo.options = gmo.options + MQC.MQGMO_WAIT;
//            gmo.waitInterval = MQC.MQWI_UNLIMITED;
            gmo.waitInterval = 20000; // 设置等待的毫秒时间限制
            //第二步：设置接收消息，correlationId设置为实时主动发送消息的时候保存的messageId
            MQMessage retrieve = new MQMessage();
            retrieve.messageId = MQC.MQMI_NONE;
            retrieve.correlationId = Hex.decodeHex(messageID.toCharArray());
            queue.get(retrieve, gmo);
            byte[] mm = new byte[retrieve.getMessageLength()];
            retrieve.readFully(mm);
            response_xml = new String(mm, "GBK");
            logger.error("receive tax response success , message content is [" + response_xml + "]");
            //第四步：保存messageId,并在发送消息的时候设置correlationId为该messageId
            queue.close();
            mqMgr.disconnect();

        } catch (JmsException jmsException) {
            logger.error("failed to receive message may be due to no message. message info is [" + jmsException.getMessage() + "]");
            return "";
        } catch (Exception e) {
            logger.error("failed to receive message. message info is [" + e.getMessage() + "]");
            return "";
        }
        return response_xml;
    }

    /**
     * 同步消息队列监听
     * 同步消息队列中有信息，则http请求银行，获取响应报文
     * 根据请求信息中的messageID，将correlationID = messageID写会到远程同步消息队列中
     * @param message
     */
    @JmsListener(destination = "${project.mq.syn-receive-queue}", selector = "JMSCorrelationID = 'ID:524551000000000000000000000000000000000000000000'")
    public void synReceive(JMSMessage message) {
        logger.error("==================== synReceive start ====================");

        try {
            CcbMessage receiveMessage = new CcbMessage();
            receiveMessage.setTypes(type_sw_yh_online_s);
            String messageID = message.getJMSMessageID().replace("ID:", "");
            receiveMessage.setMessageid(messageID);
            receiveMessage.setStatus("2");
            if (message instanceof JMSBytesMessage) {
                byte[] reqXmlBytes = new byte[(int) ((JMSBytesMessage) message).getBodyLength()];
                ((JMSBytesMessage) message).readBytes(reqXmlBytes);
                receiveMessage.setMessageSend(new String(reqXmlBytes, "GBK"));
            } else {
                JMSTextMessage textMessage = (JMSTextMessage) message;
                String msgStr = textMessage.getText();
                receiveMessage.setMessageSend(msgStr);
            }
            logger.error("syn message listener receive success , content is [" + receiveMessage.getMessageSend() + "] ");
            /**
             * 调用银行端HTTP
             *
             */
            String httpResponse ="??????????";

            receiveMessage.setMessageReceive(httpResponse);
            logger.error("request bank side , response is [" + httpResponse + "]");
            /**
             * 获取到局端信息后，将响应报文 写入同步远程队列中
             */
            this.sendMessage("syn", httpResponse, messageID);

        } catch (JMSException e) {
            logger.error("syn message listener exception [" + e.getMessage() + "]");
        } catch (UnsupportedEncodingException e) {
            logger.error("syn message listener exception [" + e.getMessage() + "]");
        } catch (Exception e) {
            logger.error("syn message listener exception [" + e.getMessage() + "]");
        }
        logger.error("==================== synReceive  end ====================");
    }

    /**
     * 异步消息队列监听
     * 异步消息队列中有信息，则http请求银行，获取响应报文
     * 根据请求信息中的messageID，将correlationID = messageID写会到远程异步消息队列中
     *
     * @param message
     */
    @JmsListener(destination = "${project.mq.asyn-receive-queue}", selector = "JMSCorrelationID = 'ID:524551000000000000000000000000000000000000000000'")
    public void asynReceive(JMSMessage message) {
        logger.error("==================== asynReceive start ====================");
        try {
            CcbMessage receiveMessage = new CcbMessage();
            receiveMessage.setTypes(type_sw_yh_batch_s);
            String messageID = message.getJMSMessageID().replace("ID:", "");
            receiveMessage.setMessageid(messageID);
            receiveMessage.setStatus("2");
            if (message instanceof JMSBytesMessage) {
                byte[] reqXmlBytes = new byte[(int) ((JMSBytesMessage) message).getBodyLength()];
                ((JMSBytesMessage) message).readBytes(reqXmlBytes);
                receiveMessage.setMessageSend(new String(reqXmlBytes, "GBK"));
            } else {
                JMSTextMessage textMessage = (JMSTextMessage) message;
                String msgStr = textMessage.getText();
                receiveMessage.setMessageSend(msgStr);
            }
            /**
             * 调用银行端HTTP
             *
             */
            logger.error("asyn message listener receive success , content is [" + receiveMessage.getMessageSend() + "] ");
            String httpResponse = "????";
            receiveMessage.setMessageReceive(httpResponse);

        } catch (JMSException e) {
            logger.error("asyn message listener exception [" + e.getMessage() + "]");
        } catch (UnsupportedEncodingException e) {
            logger.error("asyn message listener exception [" + e.getMessage() + "]");
        } catch (Exception e) {
            logger.error("asyn message listener exception [" + e.getMessage() + "]");
        }
        logger.error("==================== asynReceive  end ====================");
    }
}
